package me.jiatao.jetpack

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

val channel = Channel<Int>()

// 接受消息
suspend fun receiveEvent() {
    coroutineScope {
        println("receive thread ---> ${Thread.currentThread().name}")

        while (!channel.isClosedForReceive) {
            // receive()方法异步获取元素，如果缓冲区是空，receive() 调用者将被挂起，直到一个新值被发送到缓冲区
            // receive() 是一个挂起函数，用于同步发送方和接收方的一种机制
            val value = channel.receive()
            println("channel receive1---> $value")

            val value2 = channel.receive()
            println("channel receive2---> $value2")

            // poll()方法同步获取一个元素，如果缓冲区是空的，则返回null
            // channel.poll()
        }
    }
}

// 发送消息
suspend fun postEvent() {
    coroutineScope {
        println("send thread ---> ${Thread.currentThread().name}")

        if (!channel.isClosedForSend) {
            (1..10).forEach {
                delay(500)
                // 如果缓冲区没有满，则立即添加元素，
                // 如果缓冲区满了调用者会被挂起
                // send() 是一个挂起函数，用于同步发送方和接收方的一种机制
                channel.send(it)
                println("channel send---> $it")
                // offer()：如果缓冲区存在并且没有满立即向缓冲区添加一个元素
                // 如果添加成功会返回true, 失败会返回 false
                // channel.offer(it)
            }
        }
    }
}

/**
 * channel测试函数
 */
fun channelTest() = runBlocking {
    println("Channel test main")
    println("main thread ---> ${Thread.currentThread().name}")

    //send() 是一个挂起函数,这里使用launch，后面的代码可以继续运行
    launch {
        println("main launch1 ---> ${Thread.currentThread().name}")
        postEvent()
    }
    //测试延迟接收
    delay(1000)

    //receive() 是一个挂起函数,这里使用launch，后面的代码可以继续运行
    launch {
        println("main launch2 ---> ${Thread.currentThread().name}")
        receiveEvent()
    }
    println("after receiveEvent")

    //保持程序不退出
    delay(50000)
}


var conflatedBroadcastChannel = ConflatedBroadcastChannel<Int>()

suspend fun conflatedSend() {
    println("conflatedSend ---> ${Thread.currentThread().name}")

    if (!channel.isClosedForSend) {
        (1..10).forEach {
            delay(100)
            conflatedBroadcastChannel.send(it)
        }
    }
}

suspend fun conflatedReceive1() {
    println("conflatedReceive1 ---> ${Thread.currentThread().name}")

    val receiveChannel = conflatedBroadcastChannel.openSubscription()
    while (!receiveChannel.isClosedForReceive) {
        val value = receiveChannel.receive()
        println("conflated receive1---> $value")

        //每隔200ms取一次数据
        delay(200)
    }
}

suspend fun conflatedReceive2() {
    println("conflatedReceive2 ---> ${Thread.currentThread().name}")

    val receiveChannel = conflatedBroadcastChannel.openSubscription()
    while (!receiveChannel.isClosedForReceive) {
        val value = receiveChannel.receive()
        println("conflated receive2---> $value")

        //每隔500ms取一次数据
        delay(500)
    }

}

fun conflatedTest() = runBlocking {
    //发送数据
    launch {
        conflatedSend()
    }

    //delay(2000)

    launch {
        conflatedReceive1()
    }

    launch {
        conflatedReceive2()
    }

    delay(50000)
}


fun main() {
    conflatedTest()
}